package com.central.scheduler.api.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.central.common.entity.*;
import com.central.common.enums.PrivilegeType;
import com.central.common.enums.TaskType;
import com.central.common.enums.WriteMode;
import com.central.common.model.*;
import com.central.common.service.impl.SuperServiceImpl;
import com.central.common.utils.UserRoleUtil;
import com.central.scheduler.api.service.IExecutorService;
import com.central.scheduler.api.service.ITaskDefinitionService;
import com.central.scheduler.api.sqlparser.FlinkSqlParser;
import com.central.scheduler.api.sqlparser.HiveSqlParser;
import com.central.scheduler.api.sqlparser.MysqlSqlParser;
import com.central.scheduler.common.task.TaskTimeoutParameter;
import com.central.scheduler.common.utils.JSONUtils;
import com.central.scheduler.dao.entity.TaskNodeParam;
import com.central.scheduler.dao.mapper.*;
import com.central.scheduler.dao.model.*;
import com.central.scheduler.service.process.ProcessService;
import com.central.scheduler.common.enums.*;
import com.central.scheduler.common.utils.SnowFlakeUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.text.MessageFormat;
import java.util.*;

import static com.central.scheduler.common.Constants.MAX_TASK_TIMEOUT;


/**
 * @author Tindy
 * @date 2021/6/24
 * @describe
 */
@Slf4j
@Service
public class TaskDefinitionServiceImpl extends SuperServiceImpl<TaskDefinitionMapper, TaskDefinition> implements ITaskDefinitionService {

    @Autowired
    private ProcessDefinitionMapper processDefinitionMapper;
    @Autowired
    private ProcessDefinitionLogMapper processDefinitionLogMapper;
    @Autowired
    private ProcessTaskRelationMapper processTaskRelationMapper;
    @Autowired
    private ProcessService processService;
    @Autowired
    private IExecutorService executorService;
    @Autowired
    private WorkerGroupMapper workerGroupMapper;

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void release(TaskDefinition taskDefinition) {
        //如果该任务已经发布，则更新发布的版本
        TaskDefinition taskDefinition2 = baseMapper.selectOne(new QueryWrapper<TaskDefinition>().eq("task_defined_id", taskDefinition.getTaskDefinedId()).eq("is_tmp",false));
        if(taskDefinition2!=null){
            taskDefinition.setId(taskDefinition2.getId());
            taskDefinition.setCreateTime(taskDefinition2.getCreateTime());
        }
        setTaskOther(taskDefinition);
        saveOrUpdate(taskDefinition);
    }

    private void setTaskOther(TaskDefinition taskDefinition) {
        WorkerGroup workerGroup=null;
        switch (taskDefinition.getTaskType()){
            case SQL:
                //解析数据血缘
                sqlDataRelatedParser(taskDefinition);
                //配置worker组为SQL默认组
                workerGroup = workerGroupMapper.selectOne(new QueryWrapper<WorkerGroup>().eq("pro_id", -1).eq("task_type", TaskType.SQL));
                taskDefinition.getTaskParams().setWorkerGroupId(workerGroup.getId());
                break;
            case DATAX:
                //配置数据血缘
                dataxDataRelatedParser(taskDefinition);
                //配置worker组为DATAX默认组
                workerGroup = workerGroupMapper.selectOne(new QueryWrapper<WorkerGroup>().eq("pro_id", -1).eq("task_type", TaskType.DATAX));
                taskDefinition.getTaskParams().setWorkerGroupId(workerGroup.getId());
                break;
            case FLINK_SQL:
                //解析数据血缘
                flinkSqlDataRelatedParser(taskDefinition);
                //配置worker组为SQL默认组
                workerGroup = workerGroupMapper.selectOne(new QueryWrapper<WorkerGroup>().eq("pro_id", -1).eq("task_type", TaskType.FLINK_SQL));
                taskDefinition.getTaskParams().setWorkerGroupId(workerGroup.getId());
                break;
            default:
                break;
        }
    }

    private void flinkSqlDataRelatedParser(TaskDefinition taskDefinition) {
        FlinkSqlTaskContent taskContent = JSONUtils.parseObject(taskDefinition.getTaskContent(), FlinkSqlTaskContent.class);
        taskDefinition.setDataRelateds(FlinkSqlParser.getInstance().dataRelatedParser(taskContent));
    }

    private void dataxDataRelatedParser(TaskDefinition taskDefinition) {
        DataxTaskContent taskContent = JSONUtils.parseObject(taskDefinition.getTaskContent(), DataxTaskContent.class);
        DataRelated dataRelated = new DataRelated();
        ReaderParams reader = taskContent.getReader();
        String readerTable = (reader.getDatasourceId() + "." + reader.getDbName() + "." + reader.getTbName()).toLowerCase();
        dataRelated.getSourceTables().add(readerTable);
        Set<PrivilegeType> readerPrivilege =  new HashSet<>();
        readerPrivilege.add(PrivilegeType.SELECT);
        dataRelated.getNeedPriveleges().put(readerTable,readerPrivilege);
        WriterParams writer = taskContent.getWriter();
        String writerTable = (writer.getDatasourceId() + "." + writer.getDbName() + "." + writer.getTbName()).toLowerCase();
        dataRelated.setTargetTable(writerTable);
        Set<PrivilegeType> writerPrivilege =  new HashSet<>();
        WriteMode writeMode = writer.getWriteMode();
        switch (writeMode){
            case insert:
                writerPrivilege.add(PrivilegeType.INSERT);
                break;
            case replace:
                writerPrivilege.add(PrivilegeType.INSERT);
                writerPrivilege.add(PrivilegeType.UPDATE);
                break;
            case update:
                writerPrivilege.add(PrivilegeType.UPDATE);
                break;
            case into:
                writerPrivilege.add(PrivilegeType.INSERT);
                break;
            case overwrite:
                writerPrivilege.add(PrivilegeType.INSERT);
                break;
            default:
                log.error("writeModel 模式错误");
                throw new IllegalArgumentException("writeModel 模式错误");
        }
        dataRelated.getNeedPriveleges().put(writerTable,writerPrivilege);
        List<DataRelated> dataRelateds = new ArrayList<>();
        dataRelateds.add(dataRelated);
        taskDefinition.setDataRelateds(dataRelateds);
    }

    /**
     * 解析数据血缘
     * @param taskDefinition
     */
    private void sqlDataRelatedParser(TaskDefinition taskDefinition) {
        SqlTaskContent sqlTaskContent = JSONUtils.parseObject(taskDefinition.getTaskContent(), SqlTaskContent.class);
        switch (sqlTaskContent.getDbType()){
            case MYSQL:
                taskDefinition.setDataRelateds(MysqlSqlParser.getInstance().dataRelatedParser(sqlTaskContent));
                break;
            case HIVE:
                taskDefinition.setDataRelateds(HiveSqlParser.getInstance().dataRelatedParser(sqlTaskContent));
                break;
            default:
                throw new IllegalArgumentException("数据类型错误，数据血缘解析失败");
        }
    }


    @Override
    public PageResult<TaskDefinition> findTasks(Long proId, Map<String, Object> params) {
        Page<TaskDefinition> page = new Page<>(MapUtils.getInteger(params, "page"), MapUtils.getInteger(params, "limit"));
        List<TaskDefinition> list = baseMapper.findTasks(page, params,proId);
        long total = page.getTotal();
        return PageResult.<TaskDefinition>builder().data(list).code(0).count(total).build();
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public Result tmpRunTask(SysUser user, TaskDefinition taskDefinition) throws SnowFlakeUtils.SnowFlakeException {
        //创建临时的工作流
        ProcessDefinition processDefinition = createTmpProcess(user, taskDefinition);
        //发布执行临时工作流的命令
        return executorService.execProcessInstance(user,processDefinition.getCode(),"", CommandType.START_TMP_TASK_PROCESS, FailureStrategy.END
                ,String.valueOf(taskDefinition.getName()), TaskDependType.TASK_ONLY, WarningType.NONE,0l,RunMode.RUN_MODE_SERIAL
                ,Priority.HIGHEST,MAX_TASK_TIMEOUT,new HashMap<>(),1,0);
    }

    @Override
    public Long taskIsRelease(Long taskDefinedId, Long taskDefinedLogId) {
        if(taskDefinedId!=null){
            TaskDefinition taskDefinition = baseMapper.selectOne(new QueryWrapper<TaskDefinition>().eq("is_tmp", false).eq("task_defined_id", taskDefinedId).select("task_defined_log_id"));
            return taskDefinition!=null?taskDefinition.getTaskDefinedLogId():0l;
        }else if(taskDefinedLogId!=null){
            TaskDefinition taskDefinition = baseMapper.selectOne(new QueryWrapper<TaskDefinition>().eq("is_tmp", false).eq("task_defined_log_id", taskDefinedLogId).select("task_defined_log_id"));
            return taskDefinition!=null?taskDefinition.getTaskDefinedLogId():0l;
        }else {
            return 0l;
        }
    }

    @Override
    public Result delTask(SysUser user, Long id) {
        TaskDefinition taskDefinition = baseMapper.selectById(id);
        if(taskDefinition!=null){
            //检查操作权限  仅项目管理员与作业负责人可操作
            if(!checkOperationPermission(user,taskDefinition)) return Result.failed("操作失败，用户权限不足");
            //检查是否被非临时作业流引用
            List<ProcessDefinitionLog> processDefinitionLogs = processDefinitionLogMapper.selectList(new QueryWrapper<ProcessDefinitionLog>().eq("is_tmp",false).like("task_node_params", "%,\"taskDefinitionId\":" + id + ",%").select("name", "version"));
            if(!processDefinitionLogs.isEmpty()) return Result.failed(MessageFormat.format("操作失败，被作业流：{0} 版本{1}引用，不可删除"+processDefinitionLogs.get(0).getName(),processDefinitionLogs.get(0).getVersion()));
            baseMapper.deleteById(id);
            return Result.succeed("操作成功");
        }else return Result.failed("操作失败，作业未发布");
    }


    private ProcessDefinition createTmpProcess(SysUser user, TaskDefinition taskDefinition) throws SnowFlakeUtils.SnowFlakeException {
        //发布临时的task
        setTaskOther(taskDefinition);
        saveOrUpdate(taskDefinition);
        long code= SnowFlakeUtils.getInstance().nextId();
        String name="tmp_process_"+taskDefinition.getName()+"_"+code;
        String description=name;
        List<TaskNodeParam> taskNodeParams = new ArrayList<>();
        TaskTimeoutParameter taskTimeoutParameter = new TaskTimeoutParameter();
        taskTimeoutParameter.setFlag(TimeoutFlag.CLOSE);
        TaskNodeParam taskNodeParam = new TaskNodeParam();
        taskNodeParam.setTaskType(taskDefinition.getTaskType());
        taskNodeParam.setTaskTimeoutParameter(taskTimeoutParameter);
        taskNodeParam.setName(taskDefinition.getName());
        taskNodeParam.setTaskDefinitionId(taskDefinition.getId());
        taskNodeParams.add(taskNodeParam);
        ProcessDefinition processDefinition = new ProcessDefinition(user.getCurrentProjectId(),user.getProName(),name,code,ReleaseState.ONLINE,description,new ArrayList<>(),taskNodeParams,0,true);
        processService.saveProcessDefine(processDefinition,false);
        //create process task relation
        List<ProcessTaskRelationLog> taskRelationList=new ArrayList<>();
        ProcessTaskRelationLog processTaskRelationLog = new ProcessTaskRelationLog();
        processTaskRelationLog.setName("");
        processTaskRelationLog.setTaskType(taskDefinition.getTaskType());
        processTaskRelationLog.setProcessDefinitionVersion(processDefinition.getVersion());
        processTaskRelationLog.setProId(processDefinition.getProId());
        processTaskRelationLog.setProcessDefinitionCode(code);
        processTaskRelationLog.setPostTaskName(taskDefinition.getName());
        processTaskRelationLog.setIsTmp(true);
        taskRelationList.add(processTaskRelationLog);
        processService.saveTaskRelation(processDefinition.getProId(), processDefinition.getCode(), processDefinition.getVersion(), taskRelationList);
        return processDefinition;
    }
    /**
     * 只有项目管理员和作业流负责人拥有修改权限
     *
     * @param user
     * @param taskDefinition
     * @return
     */
    private boolean checkOperationPermission(SysUser user, TaskDefinition taskDefinition) {
        if (user.getCurrentProjectId().equals(taskDefinition.getProId()) && UserRoleUtil.checkProAdmin(user.getProRoles()))
            return true;
        if (taskDefinition.getOwners().contains(user.getUsername())) return true;
        return false;
    }
}
